from xtquant import xtdata
import sys
from datetime import datetime, timedelta
from collections import deque
import pandas as pd
from typing import Optional

from 新版2.指标计算.信号统计 import MarketMonitor
from 新版2.指标计算.指标配置 import MACDConfig
from 新版2.系统组件.基础配置 import settings
from 新版2.指标计算.成交量 import solution


def _detect_price_and_time_cols(df: pd.DataFrame) -> tuple[str, str]:
    price_col = next((c for c in ["lastPrice", "price", "close", "last", "current"] if c in df.columns), None)
    if price_col is None:
        raise ValueError("当日tick中未找到价格列")
    time_col = "time" if "time" in df.columns else next((c for c in df.columns if "time" in str(c).lower()), None)
    if time_col is None:
        raise ValueError("当日tick中未找到时间列")
    return price_col, time_col


def _ensure_time_series(df: pd.DataFrame, time_col: str) -> pd.DataFrame:
    # 与 detect_macd_signals 时间解析严格一致：毫秒时间戳直接 +8 小时
    if pd.api.types.is_integer_dtype(df[time_col]) or pd.api.types.is_float_dtype(df[time_col]):
        try:
            df = df.copy()
            df["time"] = pd.to_datetime(df[time_col], unit='ms', errors='coerce') + pd.Timedelta(hours=8)
        except Exception:
            df = df.copy()
            df["time"] = pd.to_datetime(df[time_col], unit='s', errors='coerce') + pd.Timedelta(hours=8)
    else:
        parsed = pd.to_datetime(df[time_col].astype(str), format='%Y%m%d%H%M%S', errors='coerce')
        if parsed.isna().mean() > 0.5:
            parsed = pd.to_datetime(df[time_col], errors='coerce')
        df = df.copy()
        df["time"] = parsed
    df = df[df["time"].notna()].sort_values("time").reset_index(drop=True)
    return df


def fetch_ticks_for_date(stock_code: str, trade_date: str) -> Optional[pd.DataFrame]:
    """
    获取指定交易日(YYYYMMDD)的 tick 数据（含午休过滤与价格>0过滤）。
    """
    # 先触发下载，随后读取本地缓存（xtdata 内部格式）
    xtdata.download_history_data(stock_code=stock_code, period="tick", start_time=trade_date, end_time=trade_date)
    data = xtdata.get_local_data(stock_list=[stock_code], period="tick", start_time=trade_date, end_time=trade_date)
    if not data or stock_code not in data or data[stock_code] is None or data[stock_code].empty:
        return None
    df = data[stock_code]
    price_col, time_col = _detect_price_and_time_cols(df)
    df = _ensure_time_series(df, time_col)
    # 价格过滤，与 detect_macd_signals 一致：仅保留 > 0
    df = df[df[price_col] > 0].copy()
    # 过滤午休时间
    lunch_start = pd.Timestamp("11:30:00").time()
    lunch_end = pd.Timestamp("12:59:59").time()
    df["time_only"] = df["time"].dt.time
    df = df[~((df["time_only"] >= lunch_start) & (df["time_only"] <= lunch_end))]
    # 尝试携带成交量列
    vol_col = next((c for c in ['volume', 'vol', '成交量'] if c in df.columns), None)
    cols = ["time", price_col] + ([vol_col] if vol_col else [])
    df = df[cols].rename(columns={price_col: "price", vol_col: "volume"} if vol_col else {price_col: "price"}).reset_index(drop=True)
    if 'volume' not in df.columns:
        df['volume'] = 0.0
    return df


def _to_yyyymmdd(dt: datetime) -> str:
    return dt.strftime("%Y%m%d")


def _get_latest_close_for_warmup(stock_code: str, end_date: str, days: int = 5) -> Optional[float]:
    """
    获取指定 end_date(YYYYMMDD) 之前 N 天(含当日)的日线，返回最后一个 close/price 作为预热基准。
    """
    try:
        end_dt = datetime.strptime(end_date, "%Y%m%d")
        start_dt = end_dt - timedelta(days=days)
        start_date = _to_yyyymmdd(start_dt)
        xtdata.download_history_data(stock_code=stock_code, period="1d", start_time=start_date, end_time=end_date)
        data = xtdata.get_local_data(stock_list=[stock_code], period='1d', start_time=start_date, end_time=end_date)
        if data and stock_code in data and data[stock_code] is not None and not data[stock_code].empty:
            h = data[stock_code]
            if 'close' in h.columns and not h['close'].empty:
                return float(h['close'].iloc[-1])
            if 'price' in h.columns and not h['price'].empty:
                return float(h['price'].iloc[-1])
    except Exception:
        return None
    return None


def _generate_warmup_minutes_for_date(latest_close: float, warmup_hours: int, trade_date: str) -> pd.DataFrame:
    """
    在 trade_date 当天从 08:00 开始，每5分钟生成一条 warmup K（仅用于 EMA 预热）。
    """
    try:
        end_dt = datetime.strptime(trade_date, "%Y%m%d")
        start_time = end_dt.replace(hour=8, minute=0, second=0, microsecond=0)
    except Exception:
        start_time = datetime.now().replace(hour=8, minute=0, second=0, microsecond=0)
    warmup_points = int(warmup_hours) * 12
    times = [start_time + timedelta(minutes=5 * i) for i in range(warmup_points)]
    # 小幅随机波动
    try:
        import numpy as np
        np.random.seed(42)
        std = float(latest_close) * 0.001
        changes = np.random.normal(0, std, warmup_points)
        prices = [float(latest_close) + float(x) for x in changes]
    except Exception:
        prices = [float(latest_close)] * warmup_points
    df = pd.DataFrame({
        'time': times,
        'open': prices,
        'high': [p * 1.0005 for p in prices],
        'low': [p * 0.9995 for p in prices],
        'close': prices,
    })
    return df


def backtest_intraday(stock_code: str, trade_date: Optional[str] = None):
    monitor = MarketMonitor()
    monitor.stock_codes = [stock_code]
    monitor.price_history.setdefault(stock_code, [])
    monitor.tick_counters[stock_code] = 0
    # 成交量（VMA）
    # try:
    #     monitor.solution = getattr(monitor, 'solution', None) or solution()
    # except Exception:
    #     monitor.solution = solution()

    # 交易日
    date_str = trade_date or datetime.now().strftime("%Y%m%d")

    # 拉取指定日期 tick
    df = fetch_ticks_for_date(stock_code, date_str)
    if df is None or df.empty:
        print(f"{stock_code} {date_str} 当日tick为空，终止回测")
        return

    # 预先基于当日tick计算全日MACD金叉/死叉分钟集合（带预热，与信号统计一致），避免循环内重复远程获取
    golden_minutes, death_minutes = set(), set()
    try:
        df_minutes = monitor.macd_calculator.resample_ticks_to_minutes(df, 'price')
        if df_minutes is not None and not df_minutes.empty and 'close' in df_minutes.columns:
            # 与信号统计一致：可选预热（按回测日期获取之前N天的日线）
            warmup_minutes = None
            if MACDConfig.USE_WARMUP:
                try:
                    latest_close = _get_latest_close_for_warmup(stock_code, end_date=date_str, days=MACDConfig.WARMUP_DAYS)
                    if latest_close is not None:
                        warmup_df = _generate_warmup_minutes_for_date(latest_close, MACDConfig.WARMUP_HOURS, date_str)
                        warmup_minutes = monitor.macd_calculator.resample_ticks_to_minutes(warmup_df, 'close')
                except Exception:
                    warmup_minutes = None

            macd_line, signal_line, histogram = monitor.macd_calculator.calculate_macd(
                df_minutes['close'],
                fast=MACDConfig.FAST_EMA,
                slow=MACDConfig.SLOW_EMA,
                signal=MACDConfig.SIGNAL_EMA,
                warmup_data=warmup_minutes
            )
            # 交叉判定（与实现保持一致的鲁棒容差与最小间隔）
            delta = macd_line - signal_line
            prev_delta = delta.shift(1)
            delta_change = delta - prev_delta
            eps = getattr(MACDConfig, 'CROSS_EPSILON', 1e-6)
            min_gap = max(0, int(getattr(MACDConfig, 'MIN_CROSS_GAP_BARS', 1)))

            sign_prev = (prev_delta > eps).astype(int) - (prev_delta < -eps).astype(int)
            sign_now = (delta > eps).astype(int) - (delta < -eps).astype(int)

            cross_up_mask = (
                ((sign_prev < 0) & (sign_now > 0)) |
                ((sign_prev < 0) & (sign_now == 0) & (delta_change > 0)) |
                ((sign_prev == 0) & (sign_now > 0) & (delta_change > 0))
            )
            cross_down_mask = (
                ((sign_prev > 0) & (sign_now < 0)) |
                ((sign_prev > 0) & (sign_now == 0) & (delta_change < 0)) |
                ((sign_prev == 0) & (sign_now < 0) & (delta_change < 0))
            )

            def thin_cross(mask_series, min_gap_bars):
                idx = mask_series[mask_series.fillna(False)].index.to_list()
                if not idx:
                    return mask_series
                kept = []
                last_kept = None
                for i in idx:
                    if last_kept is None or (i - last_kept) >= min_gap_bars:
                        kept.append(i)
                        last_kept = i
                new_mask = mask_series.copy()
                new_mask[:] = False
                new_mask.loc[kept] = True
                return new_mask

            if min_gap > 1:
                cross_up_mask = thin_cross(cross_up_mask, min_gap)
                cross_down_mask = thin_cross(cross_down_mask, min_gap)

            if 'time' in df_minutes.columns:
                golden_minutes = set(pd.to_datetime(df_minutes.loc[cross_up_mask.fillna(False), 'time']).dt.floor('min'))
                death_minutes = set(pd.to_datetime(df_minutes.loc[cross_down_mask.fillna(False), 'time']).dt.floor('min'))
    except Exception:
        golden_minutes, death_minutes = set(), set()

    # 本地分钟K与ATR代理，保证ATR基于“当前回放进度”的分钟K计算，避免出现全为0或与时点不匹配
    class LocalATRProxy:
        def __init__(self, lookback: int = 14, method: str = 'ema'):
            self.lookback = int(lookback)
            self.method = method
            self.minute_bars = {stock_code: deque(maxlen=1000)}  # 存 open, high, low, close, time

        def update_with_tick(self, code: str, ts: pd.Timestamp, price: float):
            minute_key = pd.to_datetime(ts).floor('min')
            bars = self.minute_bars.setdefault(code, deque(maxlen=1000))
            if not bars or bars[-1]['time'] != minute_key:
                bars.append({
                    'time': minute_key,
                    'open': price,
                    'high': price,
                    'low': price,
                    'close': price,
                })
            else:
                bar = bars[-1]
                if price > bar['high']:
                    bar['high'] = price
                if price < bar['low']:
                    bar['low'] = price
                bar['close'] = price

        def _ema(self, series: pd.Series, period: int) -> pd.Series:
            if period <= 0 or series is None or series.empty:
                return pd.Series(dtype=float)
            alpha = 2 / (period + 1)
            ema_vals = []
            last_val = None
            for val in series:
                if last_val is None:
                    last_val = val
                else:
                    last_val = alpha * val + (1 - alpha) * last_val
                ema_vals.append(last_val)
            return pd.Series(ema_vals, index=series.index)

        def calculate_atr(self, code: str, current_price: Optional[float] = None, lookback: Optional[int] = None) -> Optional[float]:
            look = int(lookback or self.lookback)
            bars = list(self.minute_bars.get(code, []))
            if len(bars) < look + 1:
                return None
            dfm = pd.DataFrame(bars)
            dfm = dfm.tail(look + 1).reset_index(drop=True)
            high = dfm['high'].astype(float)
            low = dfm['low'].astype(float)
            close = dfm['close'].astype(float)
            prev_close = close.shift(1)
            tr = pd.concat([
                (high - low),
                (high - prev_close).abs(),
                (low - prev_close).abs()
            ], axis=1).max(axis=1).iloc[1:]
            if tr.isna().any() or tr.empty or len(tr) < look:
                return None
            # 极端过滤，防止异常跳变导致失真
            if current_price is not None and any(tr > float(current_price) * 0.1):
                return None
            if self.method.lower() == 'ema':
                atr_series = self._ema(tr.reset_index(drop=True), look)
            else:
                atr_series = tr.rolling(window=look, min_periods=look).mean()
            return float(atr_series.iloc[-1]) if not atr_series.empty else None

    monitor.atr_calculator = LocalATRProxy(lookback=getattr(settings, 'ATR_LOOKBACK', 14), method='ema')

    last_action = None
    prev_minute = None
    prev_minute_last_price = None
    prev_minute_last_rsi: Optional[float] = None
    prev_rsi: Optional[float] = None
    last_cum_volume = None
    for _, row in df.iterrows():
        ts = row["time"]
        price = float(row["price"]) if row["price"] is not None else None
        if price is None:
            continue
        # 成交量：把可能的累计量换算为每tick增量，并推送至VMA
        try:
            vol_val = row.get('volume') if isinstance(row, dict) else row['volume']
            vol_f = float(vol_val) if vol_val is not None else 0.0
            if last_cum_volume is not None and vol_f >= last_cum_volume:
                per_tick_vol = vol_f - last_cum_volume
            else:
                per_tick_vol = vol_f
            last_cum_volume = vol_f
            ##monitor.solution.on_tick(stock_code, ts, per_tick_vol)
        except Exception:
            pass

        # 价格历史，用于EMA5
        try:
            monitor.price_history.setdefault(stock_code, []).append(price)
        except Exception:
            pass

        # RSI（严格同花顺口径：仅在分钟切换时，用上一分钟的收盘价更新）
        # 见下方分钟切换逻辑

        # 九转
        nine_sig = monitor.nine_turn.on_tick(stock_code, ts, price)

        # 振幅/横盘（传入最小结构的 tick 列表以复用原逻辑与打印）
        try:
            monitor.updown.updown_watch_tick(stock_code, [{"lastPrice": price}])
        except Exception:
            pass

        # MACD/RSI：仅在分钟切换时使用上一分钟的收盘触发
        curr_minute = pd.to_datetime(ts).floor('min')
        macd_sig = None
        # 更新本地分钟K（用于 ATR 计算）
        monitor.atr_calculator.update_with_tick(stock_code, ts, price)
        if prev_minute is not None and curr_minute != prev_minute:
            # 用上一分钟的最后价作为“收盘价”更新 RSI
            if prev_minute_last_price is not None:
                try:
                    monitor.rsi_calculator.update_with_close(stock_code, prev_minute_last_price)
                    # 打印“确认”RSI，时间标签按分钟结束时刻（与同花顺一致）：使用 curr_minute 作为该分钟的标签
                    rsi_confirmed = monitor.rsi_calculator.get_last_rsi(stock_code)
                    ts_min = pd.to_datetime(curr_minute).strftime('%Y-%m-%d %H:%M')
                    # if rsi_confirmed is not None:
                    #     print(f"RSI10_MIN {ts_min} {stock_code} close={prev_minute_last_price:.6f} rsi={rsi_confirmed:.2f}")
                    # else:
                    #     print(f"RSI10_MIN {ts_min} {stock_code} close={prev_minute_last_price:.6f} rsi=NA")
                except Exception:
                    pass
            if prev_minute in golden_minutes:
                macd_sig = '金叉'
                ts_str = pd.to_datetime(curr_minute).strftime('%Y-%m-%d %H:%M:%S')
                p = prev_minute_last_price if prev_minute_last_price is not None else price
                rsi_out = monitor.rsi_calculator.get_last_rsi(stock_code)
                if rsi_out is not None:
                    print(f"{ts_str} {stock_code} -> MACD 金叉 @ {p:.6f} | RSI={rsi_out:.2f}")
                else:
                    print(f"{ts_str} {stock_code} -> MACD 金叉 @ {p:.6f}")
            elif prev_minute in death_minutes:
                macd_sig = '死叉'
                ts_str = pd.to_datetime(curr_minute).strftime('%Y-%m-%d %H:%M:%S')
                p = prev_minute_last_price if prev_minute_last_price is not None else price
                print(f"{ts_str} {stock_code} -> MACD 死叉 @ {p:.6f}")
        prev_minute = curr_minute
        # 记录本分钟最后价格（用于在分钟切换时打印上一分钟的价格）
        prev_minute_last_price = price

        # 预览RSI（当前分钟，用于非切换时段展示与判定）
        try:
            rsi_preview = monitor.rsi_calculator.get_preview_rsi(stock_code, float(price))
        except Exception:
            rsi_preview = None
        if rsi_preview is None:
            rsi_preview = monitor.rsi_calculator.get_last_rsi(stock_code)

        # 用于策略判定的RSI：
        # - 若本tick触发了MACD信号（上一分钟的交叉），则使用“上一分钟确认后的RSI”（刚刚落库）
        # - 否则使用当前分钟的预览RSI
        rsi_for_decision = monitor.rsi_calculator.get_last_rsi(stock_code) if macd_sig is not None else rsi_preview

        # 买卖判定（返回 action, reasons, filters）
        action, reasons, filters = monitor.get_buy_or_sell(stock_code, price, macd_sig, rsi_for_decision, nine_sig)

        # 调试日志：VMA与过滤项细节
        # try:
        #     vstate = monitor.solution.get_state(stock_code)
        # except Exception:
        #     vstate = None
        # try:
        #     ts_min_str = pd.to_datetime(curr_minute).strftime('%Y-%m-%d %H:%M') if curr_minute is not None else str(ts)
        # except Exception:
        #     ts_min_str = str(ts)


        if action and action != last_action:
            reason_text = "; ".join(reasons) if reasons else ""
            if action.upper() == "BUY":
                print(f"{ts} {stock_code} -> {action.upper()} @ {price:.6f} | {reason_text}")
            last_action = action

    # 循环结束后，补打印最后一个分钟的确认RSI
    if prev_minute is not None and prev_minute_last_price is not None:
        try:
            monitor.rsi_calculator.update_with_close(stock_code, prev_minute_last_price)
            rsi_confirmed = monitor.rsi_calculator.get_last_rsi(stock_code)
            # 最后一根的时间标签按结束时刻：上一分钟+1分钟
            ts_min = (pd.to_datetime(prev_minute) + pd.Timedelta(minutes=1)).strftime('%Y-%m-%d %H:%M')
            # if rsi_confirmed is not None:
            #     print(f"RSI10_MIN {ts_min} {stock_code} close={prev_minute_last_price:.6f} rsi={rsi_confirmed:.2f}")
            # else:
            #     print(f"RSI10_MIN {ts_min} {stock_code} close={prev_minute_last_price:.6f} rsi=NA")
        except Exception:
            pass


if __name__ == "__main__":
    # 用法：python 回测.py [code] [YYYYMMDD]
    # default_code = settings.STOCK_CODE[0]["all_code"] if settings.STOCK_CODE else ""
    # code = sys.argv[1] if len(sys.argv) > 1 else (default_code or "513090.SH")
    # date_arg = sys.argv[2] if len(sys.argv) > 2 else None
    # if date_arg is not None and (len(date_arg) != 8 or not date_arg.isdigit()):
    #     print("无效日期格式，请使用 YYYYMMDD，例如 20250819")
    #     sys.exit(1)
    # print(f"开始当日tick回测：{code} {date_arg or datetime.now().strftime('%Y%m%d')}")
    code = "513120.SH"
    date_arg = "20250822"
    backtest_intraday(code, date_arg)